package com.shujia.flink.core;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo02WordCountBatch {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        // 设置运行的方式
        /**
         * BATCH：批处理
         * 1、底层基于MR模型
         * 2、只能用于处理有界的数据
         * STREAM：流处理
         * 1、底层基于持续流模型
         * 2、既能处理无界流也可以用于有界流
         *
         * 流批合一：在Flink中同一套DataStream的API既可以用作流处理也可以用作批处理
         */
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 加载数据
        DataStream<String> lineDS = env.readTextFile("flink/data/words.txt");

        // 切分每一行单词
        DataStream<String> wordsDS = lineDS.flatMap((line, collector) -> {
            String[] splits = line.split(",");
            // 遍历每一个单词
            for (String word : splits) {
                // 使用collector将结果数据发送给下游
                collector.collect(word);
            }
            // java中的lambda表达式有时无法自动推断类型，需要手动指定
        }, Types.STRING);
        // 4、将每个word变成word，1这种形式
        DataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1),Types.TUPLE(Types.STRING,Types.INT));

        // 5、按照每个单词进行分组
        KeyedStream<Tuple2<String, Integer>, String> grpDS = kvDS.keyBy(kv -> kv.f0);

        // 6、统计单词数量
        DataStream<Tuple2<String, Integer>> wordCntDS = grpDS.sum(1);
        wordCntDS.print();


        // 启动任务
        env.execute();


    }
}
